package com.letv.mq;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.letv.common.Constants;
import com.letv.conf.ConfigFactory;
import com.letv.dataprocess.cp_info.CpInfoDataProcessTask;
import com.letv.work.special.SpecialJob;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

/**
 * 消息队列，用于增量更新索引/CBASE
 * 
 */
public class MQLeso {
	private final static String VM_HOST =  ConfigFactory.getString("mqleso.vmhost");
	private final static String M_QUEUE_NAME =  ConfigFactory.getString("mqleso.requestQueue");
	protected final static Logger log = LoggerFactory.getLogger(MQLeso.class);
	private final static AtomicBoolean IS_BEACH = new AtomicBoolean(false); 
    // 异常尝试的次数
    private final static int EXCEPTION_MAX_NUM = 10;
    // 异常时等待时间 60秒
    private final static int EXCEPTION_LONG_WAIT_TIME = 5000;
    private static ConnectionFactory factory = null;
    
    private static void initFactory () {
        String ip = ConfigFactory.getString("mqleso.ip");
        int port = ConfigFactory.getInt("mqleso.port", 5684);
        String username = ConfigFactory.getString("mqleso.username");
        String password = ConfigFactory.getString("mqleso.password");
        factory = new ConnectionFactory();
        factory.setHost(ip);
        factory.setPort(port);
        factory.setVirtualHost(VM_HOST);
        factory.setUsername(username);
        factory.setPassword(password);
    }
	
	/**
	 * 获取MQ服务器的链接
	 */
	private static Connection connect() {
		Connection connection = null;
		try {
			connection = factory.newConnection();
			log.info("mqleso connect success");
		} catch (IOException e) {
			log.error(e.getMessage(), e);
		}
		return connection;
	}
	
	/**
	 * 接收各端消息并放到本地队列
	 */
	class MQRev implements Runnable {
		private Connection conn;
		protected Logger logger = LoggerFactory.getLogger(MQRev.class);
        Channel channel = null;
        QueueingConsumer consumer = null;
		
		protected MQRev(Connection conn) {
			this.conn = conn;
		}
		
		/**
		 * 负责接收各端mq消息
		 */
		public void doRev() {
            int exceptionNum = 0;
            boolean success = false;
		    while(true) {
		        try {
		            confirm();
                    success = true;
		            //channel.queueDeclare(M_QUEUE_NAME, false, false, false, null);
					// 从消息队列里取到下一条消息
				    if (IS_BEACH.get()) {
                        Thread.sleep(10000);
                        logger.info("Process Leso MQ Sleep,10000ms");
                        continue;
				    }
				    if (success) {
				        long mqid = System.currentTimeMillis();
				        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
				        String jsonStr = new String(delivery.getBody());
				        logger.info(mqid+",mqleso Consumer :" + jsonStr);
				        MQMessage message = new MQMessage();
				        JSONObject json = null;
				        JSONObject letvMMSMessage = null;
				        JSONObject properties = null;
				        int actionCode = 0;
				        String id = "";
				        try {
				            json =  JSON.parseObject(jsonStr);
				            letvMMSMessage = json.getJSONObject("LetvMMSMessage");
				            properties = letvMMSMessage.getJSONObject("properties");
				            id = properties.getString("id");
				            actionCode = letvMMSMessage.getInteger("messageActionCode");
				        } catch (Exception e1) {
				            logger.warn(mqid+",parse the mqleso message error! continue!", e1);
				            continue;
				        }
				        message.setId(id);
					
				        if (8881001 <= actionCode && actionCode <= 8881003) {
				            message.setType(Constants.INDEX_LESO);
				        } else if (8882001 <= actionCode && actionCode <= 8882003) {
				            message.setType(Constants.INDEX_LESO_EXTRAET);
				            actionCode += 2;
				        } else {
				            continue;
				        }
				        if (actionCode % 3 == 0) {			//修改
				            message.setAction(Constants.MQ_ACTION_MOD);
				        } else if (actionCode % 3 == 1) {	//删除
				            message.setAction(Constants.MQ_ACTION_DEL);
				        } else if (actionCode % 3 == 2) {	//添加
				            message.setAction(Constants.MQ_ACTION_ADD);
				        } else {
				            continue;
				        }
				        if (8882003 <= actionCode && actionCode <= 8882005 && actionCode % 3 == 1) {
				            message.setAction(letvMMSMessage.getString("redeliveryMode"));
				        }
				        if (message == null || StringUtils.isBlank(jsonStr)) {
				            continue;
				        }
                        if(CpInfoDataProcessTask.CP_MQ_IF_OPEN.equals("1") && actionCode == CpInfoDataProcessTask.DELETE_ALBUM_ACTION) {    // 因cp详情id取自index_leso表，故应在index_leso表将该专辑删除前读取
                            new CpInfoDataProcessTask().processIntoCbaseByAlbumId(message.getId(), actionCode);
                        }
                        // 调用增量
				        SpecialJob increme = new SpecialJob();
				        increme.fromMqMessage_leso(mqid, message.getId(), message.getAction(), message.getType());
                        if(CpInfoDataProcessTask.CP_MQ_IF_OPEN.equals("1") && actionCode == CpInfoDataProcessTask.ADD_ALBUM_ACTION) {    // 因cp详情id取自index_leso表，故应在index_leso表将该专辑添加后读取
                            new CpInfoDataProcessTask().processIntoCbaseByAlbumId(message.getId(), actionCode);
                        }
                        exceptionNum = 0;
				    }
		        } catch (Exception e) {
		            logger.error("mqleso recive error!", e);
		            try {
                        // 如果出现异常，尝试次数超过一定数量后，进后长时间尝试
                        if(exceptionNum++ > EXCEPTION_MAX_NUM) {
                            // 尝试次数过多，进入长时间等待过程
                            channel = null;
                            consumer = null;
                            Thread.sleep(EXCEPTION_LONG_WAIT_TIME);
                        } else {
                            Thread.sleep(1000);
                        }
                    } catch (InterruptedException e1) {
                        logger.error("Call doRev thread sleep exception " + e1.getMessage());
                        break;
                    }
                    success = false;
                    logger.error("Connection or doSent Error! ", e);
                }
		     }
		}
		
		public void run() {
			logger.info("rev thread starting.");
			doRev();
		}
		
	    public void confirm() throws IOException {
	        if (conn == null || ! conn.isOpen()) {
	            conn = connect();
	        }
	        if (channel == null || !channel.isOpen()){
	            channel = conn.createChannel();
	        }
	        if (consumer == null) {
	            consumer = new QueueingConsumer(channel);
	            channel.basicConsume(M_QUEUE_NAME, true, consumer);
	        }
	    }
	}
	
	public static void setIsBeach(boolean flag){
	    boolean pre = IS_BEACH.getAndSet(flag);
	    if (pre !=flag) {
	        log.info("Change the IS_BEACH OK ,the new value="+flag);
	    }
	}
	
    public static boolean getIsBeach(){
	    return IS_BEACH.get();
	}
	
	public static void start() {
        initFactory();
        for (int i=0;i<6;i++) {
            Connection conn = connect();// 建立MQ连接
            new Thread(new MQLeso().new MQRev(conn)).start();
        }
	}
	
	public static void main(String[] args) {
	}
}